package com.heima.schedule.service.impl;

import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.exception.CustomException;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.constants.ScheduleConstants;
import com.heima.model.schedule.dto.TaskDto;
import com.heima.model.schedule.entity.TaskInfo;
import com.heima.model.schedule.entity.TaskInfoLogs;
import com.heima.schedule.mapper.TaskInfoLogsMapper;
import com.heima.schedule.mapper.TaskInfoMapper;
import com.heima.schedule.service.TaskService;
import com.heima.schedule.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Resource
    private TaskInfoMapper taskInfoMapper;

    @Resource
    private TaskInfoLogsMapper taskInfoLogsMapper;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedisUtil redisUtil;

//    @Resource
//    private RedisLock redisLock;

    @Resource
    private RedissonClient redissonClient;

    /**
     * 添加任务
     * @param task
     * @return
     */
    @Override
    public long addTask(TaskDto task) {

        //参数校验
        if(task==null||task.getTaskType()==null||task.getPriority()==null||task.getExecuteTime()<=0){
            throw  new CustomException(AppHttpCodeEnum.PARAM_REQUIRE);
        }
        //添加任务到DB
        TaskInfo taskInfo=addTaskToDb(task);
        //保存任务数据到Cache
        if(taskInfo!=null){
            Boolean isLoad = addTaskToDCache(taskInfo);
            taskInfo.setIsLoad(isLoad?1:0);
            taskInfoMapper.updateById(taskInfo);
        }
        return taskInfo.getTaskId();
    }

    @Override
    public Boolean cancelTask(Long taskId){

        if(taskId==null){
            throw new CustomException(AppHttpCodeEnum.PARAM_REQUIRE);
        }
        //1.数据库中删除任务
        TaskInfo taskInfo=removeTaskFormDb(taskId);
        //2.从Redis中删除任务
        if(taskInfo!=null){
            return removeTaskFromRedis(taskInfo);
        }
        return false;

    }

    /**
     * 拉取任务消费
     * @param taskType
     * @param priority
     * @return
     */
    @Override
    public TaskInfo pollTask(Integer taskType, Integer priority) {
        //参数校验
        if(taskType==null||priority==null){
            throw  new CustomException(AppHttpCodeEnum.PARAM_REQUIRE);
        }
        //从Redis的List集合头部获取任务数据
        String taskId=pollTaskFromRedis(taskType,priority);
        if(StringUtils.isEmpty(taskId)){
            return null;
        }

        //从数据库中获取并删除任务数据
        return getAndRemoveTaskInfo(taskId);

    }

    /**
     * 每五秒 将预加载数据刷新到list中
     */
    @Scheduled(cron = "0/10 * * * * ? ")
    public void refresh(){
        log.info("定时刷新任务开始执行 , 开始尝试获取锁-----------------");
        RLock rLock = redissonClient.getLock("task_refresh_lock");
        boolean isLock = false;
        try {
            isLock = rLock.tryLock(2, 30, TimeUnit.SECONDS);
            if (!isLock) {
                log.info("获取定时刷新分布式锁失败-----------------");
                return;
            }

            log.info("获取到分布式锁,定时刷新任务开始执行----------");
            Thread.sleep(2000);
            //1. 获取所有需要刷新的集合
            Set<String> keys = redisUtil.scanKeys(ScheduleConstants.TASK_FUTURE + "*");
            if (CollectionUtils.isEmpty(keys)) {
                return;
            }
            log.info("需要刷新的集合:{}", keys);
            //2. 获取到每个集合中需要刷新的数据    task_future_1_1  ----> task_topic_1_1
            keys.stream().forEach(futureKey -> {
                //获取每个集合需要刷新的数据
                Set<String> taskIds = stringRedisTemplate.opsForZSet().rangeByScore(futureKey, 0, System.currentTimeMillis());
                if (!CollectionUtils.isEmpty(taskIds)) {
                    String topicKey = futureKey.replace(ScheduleConstants.TASK_FUTURE, ScheduleConstants.TASK_TOPIC);
                    log.info("需要刷新的任务:{},futureKey={},topicKey={}", taskIds, futureKey, topicKey);
                    //3. 从预加载ZSET集合中移除
                    //stringRedisTemplate.opsForZSet().remove(futureKey, taskIds.toArray());
                    //4. 添加任务到LIST集合
                    //stringRedisTemplate.opsForList().leftPushAll(topicKey, taskIds);
                    redisUtil.refreshWithPipeline(futureKey, topicKey, taskIds);
                }
            });
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        } finally {
            if (isLock) {
                rLock.forceUnlock();
            }
        }
    }

    /**
     * 同步任务代码
     */
    @Scheduled(fixedDelay = 10000)
    public void loadTask(){
        //1. 查询数据库中已经到了预加载时间但是没有被加载的数据
        Date preloadTime = DateUtil.offsetMinute(new Date(), 5);
        LambdaQueryWrapper<TaskInfo> wrapper = Wrappers.<TaskInfo>lambdaQuery().eq(TaskInfo::getIsLoad, 0).le(TaskInfo::getExecuteTime, preloadTime);
        List<TaskInfo> taskInfoList = taskInfoMapper.selectList(wrapper);
        log.info("查询到数据库中需要加载的任务:{}",taskInfoList.stream().map(taskInfo -> taskInfo.getTaskId()).collect(Collectors.toList()));
        if (CollectionUtils.isEmpty(taskInfoList)){
            return;
        }
        //2. 将任务数据保存到Redis
        for (TaskInfo taskInfo : taskInfoList) {
            String key = ScheduleConstants.TASK_FUTURE + taskInfo.getTaskType() + "_" + taskInfo.getPriority();
            stringRedisTemplate.opsForZSet().add(key, taskInfo.getTaskId() + "", taskInfo.getExecuteTime().getTime());
            //更新数据库加载状态
            taskInfo.setIsLoad(1);
            taskInfoMapper.updateById(taskInfo);
        }
    }


    //1.数据库根据taskId查询任务
    //删除数据库
    //修改任务日志的状态
    private TaskInfo getAndRemoveTaskInfo(String taskId) {
        //根据Id查询数据库
        TaskInfo taskInfo = taskInfoMapper.selectById(taskId);

        if(taskInfo==null){
            return null;
        }
        //数据库中删除数据
        taskInfoMapper.deleteById(taskId);
        //修改日志状态

        TaskInfoLogs taskInfoLogs = taskInfoLogsMapper.selectById(taskId);
        taskInfoLogs.setStatus(ScheduleConstants.EXECUTED);
        //更新数据库
        taskInfoLogsMapper.updateById(taskInfoLogs);
        return taskInfo;
    }

    /**
     * 从redisList 头部拉取任务
     * @param taskType
     * @param priority
     * @return
     */
    private String pollTaskFromRedis(Integer taskType, Integer priority) {

        String key=ScheduleConstants.TASK_TOPIC+taskType+"_"+priority;
        String taskId = stringRedisTemplate.opsForList().rightPop(key);
        return taskId;

    }

    private Boolean removeTaskFromRedis(TaskInfo taskInfo) {
        //获取任务执行时间
        Long executeTime = taskInfo.getExecuteTime().getTime();

        //获取任务预加载时间--当前时间之后的五分钟
        Long preLoadTime = DateUtil.offsetMinute(new Date(), 5).getTime();
        //任务执行时间小于等于当前时间，将任务数据存入到List集合排队执行
        if(executeTime<=System.currentTimeMillis()){
            String key=ScheduleConstants.TASK_TOPIC+taskInfo.getTaskType()+"_"+taskInfo.getPriority();
            stringRedisTemplate.opsForList().remove(key,1,taskInfo.getTaskId()+"");
        }else if(executeTime<=preLoadTime){
            //预加载集合key的规则:前缀_任务类型_任务优先级
            String key=ScheduleConstants.TASK_FUTURE+taskInfo.getTaskType()+"_"+taskInfo.getPriority();
            stringRedisTemplate.opsForZSet().remove(key,taskInfo.getTaskId()+"");
        }
        //任务执行时间小于等于当前时间，将任务数据存到List集合排队执行
        return true;
    }

    /**
     *
     * @return
     */
    private TaskInfo removeTaskFormDb(Long taskId) {
        TaskInfo taskInfo = taskInfoMapper.selectById(taskId);
        if(taskInfo==null){
            throw new CustomException(AppHttpCodeEnum.DATA_NOT_EXIST);
        }
      taskInfoMapper.deleteById(taskId);

        //修改task_Info_logs的任务状态

            TaskInfoLogs taskInfoLogs = taskInfoLogsMapper.selectById(taskId);
            taskInfoLogs.setStatus(ScheduleConstants.CANCELLED);
            taskInfoLogsMapper.updateById(taskInfoLogs);
            return taskInfo;

    }


    /**
     * 添加任务到DB
     * @return
     */
    private TaskInfo addTaskToDb(TaskDto task){
        //1.保存任务数据到TaskInfo表
        TaskInfo taskInfo = new TaskInfo();
        BeanUtils.copyProperties(task,taskInfo);
        taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskInfoMapper.insert(taskInfo);

        //2.保存任务数据到TaskInfoLog表
        TaskInfoLogs taskInfoLogs = new TaskInfoLogs();
        BeanUtils.copyProperties(taskInfo,taskInfoLogs);
        taskInfoLogs.setVersion(0);
        taskInfoLogs.setStatus(ScheduleConstants.SCHEDULED);
        taskInfoLogsMapper.insert(taskInfoLogs);

        return taskInfo;
    }

    /**
     * 保存任务到缓存
     */
    private Boolean addTaskToDCache(TaskInfo taskInfo){
        //获取任务执行时间
        long executeTime = taskInfo.getExecuteTime().getTime();
        //获取预加载时间(5min)
        long preloadTime = DateUtil.offsetMinute(new Date(), 5).getTime();
        //1. 任务执行时间小于等于当前时间(任务需要立刻执行) , 添加任务数据到Redis中List集合(左进右出)
        if (executeTime <= System.currentTimeMillis()) {
            //List集合 key规则 : 前缀+任务类型+优先级
            String key = ScheduleConstants.TASK_TOPIC + taskInfo.getTaskType() + "_" + taskInfo.getPriority();
            stringRedisTemplate.opsForList().leftPush(key, taskInfo.getTaskId().toString());
            return true;
        } else if (executeTime <= preloadTime) {
            //2. 任务执行时间大于当前时间小于预加载时间(任务即将执行), 添加任务数据到Redis中ZSET集合(任务数据预加载)
            //ZSET集合 key规则 : 前缀+任务类型+优先级
            String key = ScheduleConstants.TASK_FUTURE + taskInfo.getTaskType() + "_" + taskInfo.getPriority();
            stringRedisTemplate.opsForZSet().add(key, taskInfo.getTaskId().toString(), executeTime);
            return true;
        }
        return false;
    }
}
